# -*- coding: utf-8 -*-
from pykafka import KafkaClient
from pykafka.common import OffsetType
import datetime
import json

def KafkaDownloader(host_, topic_, group_id_):
    client = KafkaClient(hosts=host_)
    _topic = client.topics[bytes(topic_.encode())]
    consumer = _topic.get_simple_consumer(
        consumer_group=bytes(group_id_.encode()),
        auto_commit_enable=False,
        auto_offset_reset=OffsetType.LATEST,
        reset_offset_on_start=True
    )
    if consumer is not None:
        for message in consumer:
            if message is not None:
                yield message.value


def get_kafka_data(hosts,topic,group,interval_minutes=0):
    list_msg = []
    if interval_minutes > 0:
        end_time = str(datetime.datetime.now() + datetime.timedelta(minutes=interval_minutes))[0:16]
        for message in KafkaDownloader(hosts, topic, group):
            curr_time = str(datetime.datetime.now())[0:16]
            msg = json.loads(message.decode("UTF-8"))
            list_msg.append(msg)
            if curr_time == end_time:
                break
        return list_msg
    else:
        pass

    

if __name__ == '__main__':
    hosts = "10.10.202.13:9092"
    topic = "mb.park.in"
    group = "ywpt_monitor_park"
    interval_minutes = 1
    data = get_kafka_data(hosts, topic, group, interval_minutes)
    print(data)



